local skynet = require "skynet"
local SkynetMongo = require "skynet.db.mongo"
local Timer = require "core.Timer"

local table = table

---@class MongoAgent: Service
local M = {}

local mongo

-- { collection = {source = {}, source2 = {}, ...}}
local collectionOps = {}

-- 数据写入时间间隔，默认5分钟写入一次
local flushInterval

-- 数据写入次数间隔，默认每个集合每100次操作写入一次
local flushCount

local function runOp(mongoCollection, op)
    local ok, err, ret
    if op.args then
        ok, err, ret = mongoCollection[op.method](mongoCollection, table.unpack(op.args))
    else
        ok, err, ret = mongoCollection[op.method](mongoCollection)
    end

    if not ok then
        LLOGF("MONGO", "mongo flush error: op: %s\n ret: %s", table.dump(op), table.dump(ret))
    end
end

local function flushSource(collection, source)
    local ops = collectionOps[collection]
    if not ops then
        return
    end

    local mongoCollection = mongo[collection]
    for i = 1, #ops do
        local op = ops[i]
        if op.source == source then
            runOp(mongoCollection, op)
        end

        ops[i] = nil
    end

    collectionOps[collection] = table.clearhole(ops)
end

local function flushCollection(collection)
    local ops = collectionOps[collection]
    if not ops then
        return
    end

    local mongoCollection = mongo[collection]
    for i = 1, #ops do
        runOp(mongoCollection, ops[i])
    end

    collectionOps[collection] = nil
end

-- 修改操作入库
local function flush()
    local collections = table.keys(collectionOps)
    for i = 1, #collections do
        flushCollection(collections[i])
    end
end

local function opCollection(method, source, collection, ...)
    local args = table.pack(...)
    local op = {
        method = method,
        source = source,
    }

    if args.n > 0 then
        op.args = args
    end

    collectionOps[collection] = collectionOps[collection] or {}
    table.insert(collectionOps[collection], op)
    if #collectionOps[collection] >= flushCount then
        flushCollection(collection)
    end
end

local function formatFind(collection, query, projection, skip, limit, sort, hint)
    if type(collection) == "table" then
        return collection
    end

    return {
        collection = collection,
        query = query,
        projection = projection,
        skip = skip,
        limit = limit,
        sort = sort,
        hint = hint
    }
end

local function onDropCollection(collection)
    collectionOps[collection] = nil
end

function M.insertOne(source, collection, doc)
    opCollection("safe_insert", source, collection, doc)
end

---@param docs table<number, table> 文档列表，必须是一个数组
function M.insertMany(source, collection, docs)
    opCollection("safe_insert_many", source, collection, docs)
end

function M.deleteOne(source, collection, query)
    opCollection("safe_delete", collection, query, true)
end

function M.deleteMany(source, collection, query)
    opCollection("safe_delete", source, collection, query, false)
end

-- 对于 update 只有 {<field1> = <value1>, <field2> = <value2>, ...} 的文档，命令会替换文档
function M.updateOne(source, collection, query, update, upsert)
    opCollection("safe_update", source, collection, query, update, upsert)
end

function M.updateMany(source, collection, query, update, upsert)
    opCollection("safe_update", source, collection, query, update, upsert, true)
end

function M.upsertOne(source, collection, query, update)
    opCollection("safe_update", source, collection, query, update, true)
end

function M.replace(source, collection, query, new_doc)
    opCollection("safe_update", source, collection, query, new_doc, false)
end

function M.findMany(_, collection, query, projection, skip, limit, sort, hint)
    local args = formatFind(collection, query, projection, skip, limit, sort, hint)
    -- 查询函数直接返回
    local cursor = mongo[args.collection]:find(query, projection)
    if args.skip then
        cursor:skip(args.skip)
    end

    if args.limit then
        cursor:limit(args.limit)
    end

    if args.sort then
        cursor:sort(args.sort)
    end

    if args.hint then
        cursor:hint(args.hint)
    end

    local ret = {}
    while cursor:hasNext() do
        table.insert(ret, cursor:next())
    end

    cursor:close()
    return ret
end

function M.findOne(_, collection, query, projection)
    return mongo[collection]:findOne(query, projection)
end

-- 执行查询的命令，会立即执行并且并且返回结果
function M.runQueryCmd(_, collection, cmd, cmd_v, ...)
    --[[
        这里只 flush 了当前连接的 collection_op
        使用时要注意，如果一个集合会在多个地方进行更新，那么会导致查询不正确，
        如果查询可以在 uniqueservice 中处理(原则应该都是可以的，在启动时查询，然后维护状态即可)，
        最好是在对应的 uniqueservice 中进行数据处理，而不是使用 mongodb
    ]]
    flushCollection(collection)
    return mongo[collection]:runCommand(cmd, cmd_v, ...)
end

-- 执行修改的命令，命令会写入 collection_ops 中，后续一起进行更新
function M.runModifyCmd(source, collection, cmd, cmd_v, ...)
    opCollection("runCommand", source, collection, cmd, cmd_v, ...)
end

function M.count(_, collection, query, projection, skip, sort, hint)
    local args = formatFind(collection, query, projection, skip, nil, sort, hint)
    -- 查询函数直接返回
    local cursor = mongo[args.collection]:find(query, projection)
    if args.skip then
        cursor:skip(args.skip)
    end

    if args.sort then
        cursor:sort(args.sort)
    end

    if args.hint then
        cursor:hint(args.hint)
    end

    return cursor:count()
end

function M.dropCollection(_, collection)
    -- 这里没有同步到其他 MongoAgent，设计时注意一个通用的 collection 应该是一个实例操作即可
    onDropCollection(collection)
    mongo[collection]:drop()
end

-- 聚合，直接返回
function M.aggregate(_, collection, pipeline, options)
    return mongo[collection]:aggregate(pipeline, options)
end

function M.flush()
    local collections = table.keys(collectionOps)
    for i = 1, #collections do
        flushCollection(collections[i])
    end
end

function M.flushCollection(_, collection)
    flushCollection(collection)
end

function M.flushSource(source)
    flushSource(source)
end

function M.start(_, conf)
    assert(conf.host and conf.port, "host and port must be set")
    local client = SkynetMongo.client(conf)

    flushInterval = conf.flush_interval or 100
    flushCount = conf.flush_count or 100
    mongo = client[conf.db]
    if conf.username then
        mongo:auth(conf.username, conf.password)
    end

    Timer.intervalSec(flush, flushInterval, flushInterval)
    return true
end

function M.init()
    skynet.dispatch_src_lua(M)
end

return M
